package com.liu.webstore.event;

import com.alibaba.fastjson.JSONObject;
import com.liu.webstore.common.Constant;
import com.liu.webstore.entity.DiscussPost;
import com.liu.webstore.entity.Event;
import com.liu.webstore.entity.Message;
import com.liu.webstore.service.DiscussPostService;
import com.liu.webstore.service.ElasticsearchService;
import com.liu.webstore.service.MessageService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

/**
 * @Author LiuYumei
 * @Date 2022/8/13 10:35
 */
@Component
public class EventConsumer {
    private static final Logger log = LoggerFactory.getLogger(EventConsumer.class);

    @Autowired
    private MessageService messageService;

    @Autowired
    private DiscussPostService discussPostService;

    @Autowired
    private ElasticsearchService elasticsearchService;

    @KafkaListener(topics = {Constant.TOPIC_COMMENT, Constant.TOPIC_LIKE, Constant.TOPIC_FOLLOW})
    public void handleMessage(ConsumerRecord record) {
        Event event = judge(record);
        if (event == null) {
            return;
        }
        // 发送站内信
        Message message = new Message();
        message.setFromId(Constant.SYSTEM_USER_ID);
        message.setToId(event.getEntityUserId());
        message.setConversationId(event.getTopic());
        message.setCreateTime(new Date());

        Map<String, Object> comment = new HashMap<>();
        comment.put("userId", event.getUserId());
        comment.put("entityType", event.getEntityType());
        comment.put("entityId", event.getEntityId());
        if (Objects.nonNull(event.getData())) {
            comment.putAll(event.getData());
        }
        message.setContent(JSONObject.toJSONString(comment));
        messageService.saveMessage(message);
    }

    /**
     * 消费发帖事件
     */
    @KafkaListener(topics = {Constant.TOPIC_PUBLISH})
    public void handlePublishMessage(ConsumerRecord record) {
        Event event = judge(record);
        if (event == null) {
            return;
        }

        DiscussPost discussPost = discussPostService.findDiscussPost(event.getEntityId());
        elasticsearchService.saveDiscussPost(discussPost);
    }

    /**
     * 消费删帖事件
     */
    @KafkaListener(topics = {Constant.TOPIC_DELETE})
    public void handleDeleteMessage(ConsumerRecord record) {
        Event event = judge(record);
        if (event == null) {
            return;
        }
        elasticsearchService.deleteDiscussPostById(event.getEntityId());
    }

    @Nullable
    private Event judge(ConsumerRecord record) {
        if (Objects.isNull(record)) {
            log.error("消息内容为空！");
            return null;
        }
        Event event = JSONObject.parseObject(record.value().toString(), Event.class);
        if (Objects.isNull(event)) {
            log.error("消息格式错误！");
            return null;
        }
        return event;
    }
}
